Methodology

SELECT
-- извлекаем данные из JSON:
	JSONExtractString(_airbyte_data, 'event_receive_datetime') AS __date, 
	JSONExtractString(_airbyte_data, '__clientName') AS __clientName, 
	JSONExtractString(_airbyte_data, '__productName') AS __productName, 
	JSONExtractString(_airbyte_data, 'appmetrica_device_id') AS appmetrica_device_id, 
	JSONExtractString(_airbyte_data, 'city') AS city, 
	JSONExtractString(_airbyte_data, 'deeplink_url_parameters') AS deeplink_url_parameters, 
	JSONExtractString(_airbyte_data, 'event_receive_datetime') AS event_receive_datetime, 
	JSONExtractString(_airbyte_data, 'google_aid') AS google_aid, 
	JSONExtractString(_airbyte_data, 'ios_ifa') AS ios_ifa, 
	JSONExtractString(_airbyte_data, 'os_name') AS os_name, 
	JSONExtractString(_airbyte_data, 'profile_id') AS profile_id, 
	JSONExtractString(_airbyte_data, 'publisher_name') AS publisher_name,
-- добавляем технические столбцы:
	toLowCardinality(_dbt_source_relation) AS __table_name, -- столбец с названием таблицы, содержащей сырые данные 
	toDateTime32(substring(toString(_airbyte_extracted_at), 1, 19)) AS __emitted_at, -- столбец с датой выгрузки данных
	NOW() AS __normalized_at -- когда была проведена нормализация
-- данные берём из таблицы, которую Airbyte передал в хранилище данных:
FROM ((
	SELECT 
		toLowCardinality(
			'datacraft_clientname_raw__stream_appmetrica_default_accountid_deeplinks'
						) AS _dbt_source_relation,
		toString("_airbyte_raw_id") AS _airbyte_raw_id,
		toString("_airbyte_data") AS _airbyte_data,
		toString("_airbyte_extracted_at") AS _airbyte_extracted_at
	FROM airbyte_internal.datacraft_clientname_raw__stream_appmetrica_default_accountid_deeplinks
	))

{{ datacraft.normalize() }}

{{ config(
        materialized='table',
        order_by=('__table_name'),
        on_schema_change='fail'
	    ) 
}}

SELECT * FROM {{ ref('normalize_appmetrica_registry_default_profiles') }}

{{ config(
        materialized='incremental',
        order_by=('__date', '__table_name'),
        incremental_strategy='delete+insert',
        unique_key=['__date', '__table_name'],
        on_schema_change='fail'
	    ) 
}}

SELECT * FROM {{ ref('normalize_appmetrica_events_default_installations') }}

{{ datacraft.incremental() }}

{{ config(
    materialized='incremental',
    order_by=('__date', '__table_name'),
    incremental_strategy='delete+insert',
    unique_key=['__date', '__table_name'],
    on_schema_change='fail'
) }}

-- для каждого стрима создаём его CTE с одинаковым набором полей и их расположением 
-- первый стрим - deeplinks
WITH join_appmetrica_events_deeplinks AS (
SELECT
    toDateTime(__date) AS __date, 
    toLowCardinality(__table_name) AS __table_name,
    toDateTime(event_datetime) AS event_datetime,
    toLowCardinality(splitByChar('_', __table_name)[6]) AS accountName,
    appmetrica_device_id AS appmetricaDeviceId,
    assumeNotNull(COALESCE(nullIf(google_aid, ''), nullIf(ios_ifa, ''), appmetrica_device_id, '')) AS mobileAdsId,
    profile_id AS crmUserId,
    '' AS visitId,
    '' AS clientId,
    '' AS promoCode,
    os_name AS osName,
    city AS cityName,
    assumeNotNull(coalesce({{ datacraft.get_adsourcedirty() }}, publisher_name, '')) AS adSourceDirty,
<...>
    0 AS installs,
    '' AS installationDeviceId,
    __emitted_at,
    toLowCardinality('AppDeeplinkStat') AS __link
FROM {{ ref('incremental_appmetrica_events_default_deeplinks') }}
)

<...>

-- пятый стрим - sessions_starts
, join_appmetrica_events_sessions_starts AS (
SELECT
   <...>
FROM {{ ref('incremental_appmetrica_events_default_sessions_starts') }}
)

-- теперь делаем  UNION записанных ранее CTE
, final_union AS (
SELECT * 
FROM join_appmetrica_events_deeplinks
UNION ALL
SELECT * 
FROM join_appmetrica_events_events
UNION ALL
SELECT * 
FROM join_appmetrica_events_install
UNION ALL
SELECT * 
FROM join_appmetrica_events_screen_view
UNION ALL
SELECT * 
FROM join_appmetrica_events_sessions_starts
)

SELECT *
FROM final_union

{{ datacraft.join() }}

{{ config(
      materialized='incremental',
      order_by=('__date', '__table_name'),
      incremental_strategy='delete+insert',
      unique_key=['__date', '__table_name'],
      on_schema_change='fail'
		) 
}}

WITH all_events AS (
SELECT *
FROM {{ ref('join_appmetrica_events') }}

UNION ALL

SELECT *
FROM {{ ref('join_ym_events') }}
)

SELECT *
FROM all_events

{{ config(
      materialized='table',
      order_by=('__table_name'),
      on_schema_change='fail'
		 ) 
}}

SELECT * FROM {{ ref(join_источник_registry_линк) }}

{{ config(
      materialized='table',
      order_by=('__table_name'),
      on_schema_change='fail'
		 ) 
}}

SELECT * FROM {{ ref(join_источник_registry_линк) }}
UNION ALL
SELECT * FROM {{ ref(join_другойисточник_registry_линк) }}

{{ datacraft.combine() }}

{{ config(
    materialized='table',
    order_by=('__table_name'),
    on_schema_change='fail'
		) 
}}

{% set metadata = fromyaml(metadata()) %}

SELECT
    *, 
    {{ link_hash('MediaplanStat', metadata) }},
    {{ entity_hash('UtmHash', metadata) }}

FROM {{ ref('combine_periodstat') }}

{{ datacraft.hash() }} 

{{ config(
    materialized='table'
		) 
}}

SELECT MAX(accountName) AS accountName,
       <...>
       SUM(adCost) AS adCost,
       SUM(impressions) AS impressions,
       SUM(clicks) AS clicks
FROM {{  ref('hash_datestat') }}
GROUP BY __id

{{ datacraft.link() }} 

{{
    config(
        materialized = 'table',
        order_by = ('__datetime')
    )
}}

SELECT 
* 
FROM  {{ ref('link_events') }} 

{{
    config(
        materialized = 'table',
        order_by = ('__datetime')
    )
}}

WITH link_events AS (SELECT * 
FROM  {{ ref('link_events') }}),

link_registry AS (SELECT * 
FROM  {{ ref('link_registry') }})

SELECT * FROM link_events as t1
LEFT JOIN link_registry as t2 USING(utmHashHash)

{{ datacraft.full() }} 

{{ config(
        materialized='table',
        on_schema_change='fail'
        ) 
}}
select
        tuple(toLowCardinality(__link), __datetime,  __id) as hash,
        tuple(toLowCardinality('CrmUserHash'), toDateTime(0),  CrmUserHash) as node_left
from {{ ref('link_events') }}
where nullIf(CrmUserHash, '') is not null

union all
        
select
      tuple(toLowCardinality(__link), __datetime,  __id) as hash,
      tuple(toLowCardinality('YmClientHash'), toDateTime(0),  YmClientHash) as node_left
from {{ ref('link_events') }}
where nullIf(YmClientHash, '') is not null
            
union all
<...>

{{
    config(
        materialized='table',
        order_by=('key_number')
    )
}}

with all_keys as
(
    select distinct hash as key_hash from {{ ref('graph_tuples') }}
    union distinct select distinct node_left as key_hash from {{ ref('graph_tuples') }}
)

select *, row_number() over() as key_number from all_keys

{{
    config(
        materialized='table',
        order_by=('key_hash')
    )
}}

select * from {{ ref('graph_lookup') }}

{{
    config(
        materialized='table',  
        post_hook= {
            'sql': 'insert into {{target.schema}}.graph_edge(node_id_left, node_id_right, group_id, has_changed)
                    select
                        node_id_right,
                        node_id_left,
                        group_id,
                        has_changed
                    from {{target.schema}}.graph_edge;'
        }
    )
}}

with join_left as (
    select key_number as node_id_left, 
		   node_left
    from {{ ref('graph_tuples') }} x
    join {{ ref('graph_unique') }} y on x.hash = y.key_hash
)

select node_id_left, 
       key_number as node_id_right, 
       node_id_left as group_id, 
       1 as has_changed 
from join_left x
join {{ ref(graph_unique) }} y on x.node_left = y.key_hash

{{
    config(
        materialized='table',
        order_by=('node_id_left'),
        pre_hook="{{ datacraft.calc_graph() }}"
    )
}}

select 
    node_id_left,
    min(group_id) as qid
from {{ ref('graph_edge') }}
group by node_id_left

Код макроса  `datacraft.calc_graph()` выглядит следующим образом:

{% macro calc_graph() %}
    {# Запрос для обновления правой таблицы #}
    {% set right_query %}
        create or replace table {{ target.schema }}.graph_right engine=Log() as
        with
        min_group_id as (
            select
                node_id_left,
                min(group_id) as min_group_id
            from {{ target.schema }}.graph_edge
            group by node_id_left
        )

        select
            node_id_left,
            node_id_right,
            min_group_id as group_id,
            min_group_id != e.group_id as has_changed
        from {{ target.schema }}.graph_edge e
        join min_group_id r on r.node_id_left = e.node_id_left
    {% endset %}

    {# Запрос для обновления левой таблицы #}
    {% set left_query %}
        create or replace table {{ target.schema }}.graph_edge engine=Log() as
        with
        min_group_id as (
            select
                node_id_right,
                min(group_id) as min_group_id
            from {{ target.schema }}.graph_right
            group by node_id_right
        )

        select
            node_id_left,
            node_id_right,
            min_group_id as group_id,
            min_group_id != e.group_id as has_changed
        from {{ target.schema }}.graph_right e
        join min_group_id r on r.node_id_right = e.node_id_right
    {% endset %}

    {# Запрос для проверки наличия изменений #}
    {% set check_changed %}
        select 
            max(has_changed) 
        from {{ target.schema }}.graph_edge
    {% endset %}

    {# Если необходимо выполнить запросы #}
    {% if execute %}
        {% set ns = namespace(check_change=1) %}
        {% for i in range(0, 14) %}
            {{ log("Running iteration " ~ i) }}
            {{ check_change }}

            {# Проверяем, были ли изменения #}
            {% if ns.check_change == 1 %}
                {# Обновляем правую таблицу #}
                {% do run_query(right_query) %}
                {# Обновляем левую таблицу #}
                {% do run_query(left_query) %}
                {# Проверяем наличие изменений в данных #}
                {% set ns.check_change = run_query(check_changed).rows[0][0] %}
                {{ log('VALUE: ' ~ ns.check_change) }}
            {% endif %}
        {% endfor %}
    {% endif %}
{% endmacro %}

{{
    config(
        materialized='table', 
        order_by=('__datetime', '__link', '__id'), 
        pre_hook="{{ datacraft.calc_graph() }}"  
    )
}}

select  
    toLowCardinality(tupleElement(key_hash, 1)) as __link,
    tupleElement(key_hash, 2) as __datetime,
    tupleElement(key_hash, 3) as __id,
    qid
from {{ ref('graph_glue') }} 
join {{ ref('graph_lookup') }} on key_number = node_id_left 

{{
    config(
        materialized='table',  
        order_by=('qid', '__datetime','__link','__id')  
    )
}}

select
    y.qid, x.*
from {{ ref('full_events') }} as x
left join {{ ref('graph_qid') }} as y
    using (__datetime,__link, __id)

{{
    config(
        materialized='table', 
        order_by=('qid', '__datetime','__link','__id')  
    )
}}

select
    qid, 
    __link,
    CASE
     WHEN __link = 'VisitStat'  and osName = 'web'  THEN  1  
     WHEN __link = 'AppInstallStat'  and installs >= 1  THEN  2    
     WHEN __link = 'AppSessionStat'  and sessions >= 1  THEN  3
     WHEN __link = 'AppDeeplinkStat'  THEN  3    
     WHEN __link = 'AppEventStat'  and screenView >= 1  THEN  4
     ELSE 0
    END as __priority, 
    __id,
    __datetime,
    toLowCardinality(CASE
                         WHEN __link = 'VisitStat' THEN 'visits_step' 
                         WHEN __link = 'AppInstallStat' THEN 'install_step'    
                         WHEN __link = 'AppSessionStat' THEN 'app_visits_step'
                            WHEN __link = 'AppDeeplinkStat' THEN 'app_visits_step'
                         WHEN __link = 'AppEventStat' THEN 'event_step'
                     END) as __step
from {{ ref('attr_myfirstfunnel_prepare_with_qid') }}

{{
    config(
        materialized='table',
        order_by=('qid', '__datetime', '__link', '__id')
    )
}}

select
    *,
    row_number() over (partition by qid order by __datetime, __priority, __id) AS __rn
from {{ ref('attr_myfirstfunnel_create_events') }}

{{
    config(
        materialized='table',
        order_by=('qid', '__datetime', '__link', '__id')
    )
}}

with prep_new_period as (
    select *,
           max(case when __priority in [1, 2, 3, 4] then __datetime else null end) 
                       over (partition by qid order by __rn rows between unbounded preceding and 1 preceding) as prep_new_period
    from {{ ref('attr_myfirstfunnel_add_row_number') }}
)

select
    qid, 
    __link,
    __priority,
    __id,
    __datetime,
    __rn,
    __step,
    CASE
     WHEN __link = 'VisitStat' and toDate(__datetime) - toDate(prep_new_period) < 90  THEN false
     WHEN __link = 'AppInstallStat' and toDate(__datetime) - toDate(prep_new_period) < 30  THEN false
     WHEN __link = 'AppSessionStat' and toDate(__datetime) - toDate(prep_new_period) < 30  THEN false
     WHEN __link = 'AppDeeplinkStat' and toDate(__datetime) - toDate(prep_new_period) < 30  THEN false
     WHEN __link = 'AppEventStat' and toDate(__datetime) - toDate(prep_new_period) < 7  THEN false
     ELSE true
    END as __is_new_period
from prep_new_period  

{{
    config(
        materialized='table',
        order_by=('qid', '__datetime', '__link', '__id')
    )
}}

select *,
       sum(toInt32(__is_new_period)) over (partition by qid order by __rn) AS __period_number
from {{ ref('attr_myfirstfunnel_find_new_period') }}

{{
    config(
        materialized='table',
        order_by=('qid', '__datetime','__link','__id')
    )
}}

with calc_max_priority as (
    select
        qid, 
        __link,
        __id,
        __datetime,
        __rn,
        __priority,
        __period_number,
        __step,
        max(__priority) over(partition by qid) as max_priority
    from {{ ref('attr_myfirstfunnel_calculate_period_number') }}
),

generate_all_priorities as (
    select
        distinct qid, __link,
        arrayJoin(range(1, assumeNotNull(max_priority) + 1)) as gen_priority
    from calc_max_priority
),

final as (
    select
        first_value(__id) OVER (PARTITION BY qid ORDER BY gen_priority ROWS BETWEEN current row AND UNBOUNDED FOLLOWING ) as __id,
        gen_priority as __priority,
        qid, __link,
        first_value(__datetime) OVER (PARTITION BY qid ORDER BY gen_priority ROWS BETWEEN current row AND UNBOUNDED FOLLOWING ) as __datetime,
        first_value(__period_number) OVER (PARTITION BY qid ORDER BY gen_priority ROWS BETWEEN current row AND UNBOUNDED FOLLOWING ) as __period_number,
        case when calc_max_priority.qid = 0 then true else false end as __if_missed,
        __step
    from generate_all_priorities
    left join calc_max_priority
        on generate_all_priorities.qid = calc_max_priority.qid and
           generate_all_priorities.gen_priority = calc_max_priority.__priority
)

select
    qid, __link, __id,
    __priority, __datetime,
    __period_number,
    __if_missed,__step,
    row_number() over (partition by qid order by __datetime, __priority, __id) AS __rn
from final

{{
    config(
        materialized='table',
        order_by=('qid', '__period_number', '__datetime', '__priority', '__id')
    )
}}

select 
    y.__period_number as __period_number, 
    y.__if_missed as __if_missed, 
    y.__priority as __priority, 
    y.__step as __step,
    x.*EXCEPT(adSourceDirty),
    CASE
     WHEN LENGTH (adSourceDirty) < 2 THEN 1
     WHEN match(adSourceDirty, 'Органическая установка') THEN 2
     WHEN __priority = 4 and not __if_missed = 1 THEN 3
     WHEN __priority = 3 and not __if_missed = 1 THEN 4
     WHEN __priority = 2 and not __if_missed = 1 THEN 5
     WHEN __priority = 1 and not __if_missed = 1 THEN 6
     ELSE 0
    END as __last_click_rank,
    CASE
     WHEN __priority = 3 and not __if_missed = 1 THEN 1
     WHEN __priority = 2 and not __if_missed = 1 THEN 2
     WHEN __priority = 1 and not __if_missed = 1 THEN 3
     ELSE 0
    END as __first_click_rank,
    CASE
     WHEN  __if_missed and __priority = 1 THEN '[Без веб сессии]'
     WHEN  __if_missed and __priority = 2 THEN '[Без установки]'
     WHEN  __if_missed and __priority = 3 THEN '[Без апп сессии]'
     WHEN  __if_missed and __priority = 4 THEN ''
     ELSE adSourceDirty
    END as adSourceDirty
from {{ ref('attr_myfirstfunnel_prepare_with_qid') }} AS x
join {{ ref('attr_myfirstfunnel_create_missed_steps') }} AS y
    using (qid, __datetime, __link, __id)

{{
    config(
        materialized='table',
        order_by = ('qid', '__datetime', '__id')
    )
}}

with
max_last_click_rank as (
    select *
        ,max(__last_click_rank) over(partition by qid, __period_number order by __datetime, __priority, __id) as __max_last_click_rank
        ,max(__first_click_rank) over(partition by qid, __period_number order by __datetime, __priority, __id) as __max_first_click_rank
     from {{ ref('attr_myfirstfunnel_join_to_attr_prepare_with_qid') }}
                       ),
target_count as (
    select *
         , __last_click_rank = __max_last_click_rank as  __last_click__rank_condition
         , sum(case when __last_click__rank_condition then 1 else 0 end) over(partition
         by qid, __period_number order by __datetime, __priority, __id) as
         __last_click__target_count 
         , __first_click_rank = __max_first_click_rank as  __first_click__rank_condition
         , sum(case when __first_click__rank_condition then 1 else 0 end) over(partition
         by qid, __period_number order by __datetime, __priority, __id) as
         __first_click__target_count 
	from max_last_click_rank
)

SELECT qid, __datetime, __id, __priority,`__if_missed`, __link, __period_number
            , first_value(utmSource) over(partition by qid, __period_number, __last_click__target_count  order by  __datetime, __priority, __id) as __myfirstfunnel_last_click_utmSource
            , first_value(utmMedium) over(partition by qid, __period_number, __last_click__target_count  order by  __datetime, __priority, __id) as __myfirstfunnel_last_click_utmMedium
            , first_value(utmCampaign) over(partition by qid, __period_number, __last_click__target_count  order by  __datetime, __priority, __id) as __myfirstfunnel_last_click_utmCampaign
            , first_value(utmTerm) over(partition by qid, __period_number, __last_click__target_count  order by  __datetime, __priority, __id) as __myfirstfunnel_last_click_utmTerm        
            , first_value(utmContent) over(partition by qid, __period_number, __last_click__target_count  order by  __datetime, __priority, __id) as __myfirstfunnel_last_click_utmContent        
            , first_value(adSourceDirty) over(partition by qid, __period_number, __last_click__target_count  order by  __datetime, __priority, __id) as __myfirstfunnel_last_click_adSourceDirty
<...>
FROM target_count

{{
    config(
        materialized = 'table',
        order_by = ('__datetime')
    )
}}

with out as ( 
        select * except(_dbt_source_relation) 
        from  {{ ref('ttr_myfirstfunnel_join_to_attr_prepare_with_qid') }}
        join  {{ ref(attr_myfirstfunnel_model) }}
            using (qid, __datetime, __id, __link, __period_number, __if_missed,
            __priority)
    )  
select * from out

{{
    config(
        materialized = 'table',
        order_by = ('__datetime')
    )
}}

SELECT * FROM (
        (
            select toLowCardinality('full_datestat')  as _dbt_source_relation,
                   toString("None") as None ,
                   toDate("__date") as __date ,
                   toString("reportType") as reportType ,
                   toString("accountName") as accountName ,
                   toString("__table_name") as __table_name ,
                   toString("adSourceDirty") as adSourceDirty ,
                   <...>
                   toString('') as __myfirstfunnel_first_click_utmTerm ,
                   toString('') as __myfirstfunnel_first_click_utmContent ,
                   toString('') as __myfirstfunnel_first_click_adSourceDirty 
            from {{ ref('full_datestat') }}
        )

        union all

        (
            select toLowCardinality('attr_myfirstfunnel_final_table')  as _dbt_source_relation,
                   toString("None") as None ,
                   toDate("__date") as __date ,
                   toString('') as reportType ,
                   toString("accountName") as accountName ,
                   toString("__table_name") as __table_name ,
                   toString("adSourceDirty") as adSourceDirty ,
                   <...>
                   toString("__myfirstfunnel_first_click_utmTerm") as __myfirstfunnel_first_click_utmTerm ,
                   toString("__myfirstfunnel_first_click_utmContent") as __myfirstfunnel_first_click_utmContent ,
                   toString("__myfirstfunnel_first_click_adSourceDirty") as __myfirstfunnel_first_click_adSourceDirty 
            from {{ ref('attr_myfirstfunnel_final_table') }}
        )

        ) 
    WHERE 
    splitByChar('_', __table_name)[4] = 'yd'
    and 
    splitByChar('_', __table_name)[4] = 'testaccount'
    and 
    splitByChar('_', __table_name)[4] = 'default'
    UNION ALL
<...>